import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
// import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class HelloWorld{
    public static void main(String[] args) throws Exception {
        System.out.println("-----------start!");
        // 1. 初始化Flink执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 初始化TableEnvironment
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // Define the source table from CSV file
        String csvSourceDDL =
                "CREATE TABLE csv_source (\n" +
                        "    id STRING,\n" +
                        "    name STRING,\n" +
                        "    age STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'filesystem',\n" +
                        "    'path' = '/yourpath/csvfile.csv',\n" + // Replace with your actual path
                        "    'format' = 'csv',\n" +
                        "    'csv.field-delimiter' = ',',\n" + // Set the field delimiter if necessary
                        "    'csv.line-delimiter' = '\\n'\n" + // Set the line delimiter if necessary
                        ")";
        tEnv.executeSql(csvSourceDDL);

        System.out.println("-----------add source!");

        // TableResult result = tEnv.executeSql("SELECT id, name, age FROM csv_source");
        // result.print();

        // System.out.println("-----------print!");

        // Define the sink table to MySQL
        String mysqlSinkDDL =
                "CREATE TABLE mysql_sink (\n" +
                        "    id STRING PRIMARY KEY,\n" +
                        "    name STRING,\n" +
                        "    age STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'jdbc',\n" +
                        "    'url' = 'jdbc:mysql://localhost:3306/sink_db',\n" + // Replace with your actual DB URL
                        "    'table-name' = 'sind_table',\n" + // Replace with your actual table name
                        "    'username' = 'username',\n" + // Replace with your actual username
                        "    'password' = 'passwrd'\n" + // Replace with your actual password
                        // "    'driver' = 'com.mysql.cj.jdbc.Driver'\n" + // JDBC driver name
                        ")";
        tEnv.executeSql(mysqlSinkDDL);

        System.out.println("-----------add sink!");

        // Insert data from source to sink
        tEnv.executeSql("INSERT INTO mysql_sink SELECT * FROM csv_source");

        System.out.println("-----------insert data!");

        // Execute the job
        //env.execute("Flink CSV to MySQL Job");
        System.out.println("---------end!");
    }
}